package com.example.demo.rocketmq;

import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.exception.MQClientException;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

/**
 * <p>Description: mq消费者</p>
 * <p>Date: 2020/7/9 15:56</p>
 *
 * @author Eric Lee
 * @version v1.0.0
 **/

@Configuration
@EnableConfigurationProperties({RocketMQProperties.class})
@ConditionalOnProperty(prefix = "rocketmq", name = "isEnable", havingValue = "true")
@Slf4j
public class RocketMQConsumerConfig {

    @Autowired
    private RocketMsgListener msgListener;

    @Autowired
    private RocketMQProperties rocketMQProperties;

    @Bean
    public DefaultMQPushConsumer getRocketMQConsumer() {
        String groupName = rocketMQProperties.getConsumer().getGroupName();
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(groupName);
        consumer.setNamesrvAddr(rocketMQProperties.getConsumer().getNamesrvAddr());
        consumer.setConsumeThreadMin(rocketMQProperties.getConsumer().getConsumeThreadMin());
        consumer.setConsumeThreadMax(rocketMQProperties.getConsumer().getConsumeThreadMax());
        consumer.registerMessageListener(msgListener);
        try {
            String topic = rocketMQProperties.getConsumer().getTopic();
            String tag = rocketMQProperties.getConsumer().getTag();
            consumer.subscribe(topic, tag);
            log.info("================>消费者创建完成，ConsumerGroupName:{}<================", groupName);
            consumer.start();
            log.info("============>消费者监听开始,groupName:{},topic:{}<============", groupName, topic);
        } catch (MQClientException e) {
            log.error("===消费者启动失败=== msg:" + e.getMessage(), e);
        }
        return consumer;
    }

}
